--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
Commit 7a419f96ee5d1aa05e48e3d13dd4f7a70c0503ef
Parents : b8ef3d1
Author : Sudo-Ivan <ivan@quad4.io>
Signature : Signature validation error
Date : 2026-01-08T19:29:30-06:00
feat(tests): add unit tests for auto propagation API and logic
Changes
3 files changed, 211 insertions(+), 0 deletions(-)
Diff
diff --git a/tests/backend/test_auto_propagation.py b/tests/backend/test_auto_propagation.py
new file mode 100644
index 00000000..6eb6cc99
--- /dev/null
+++ b/tests/backend/test_auto_propagation.py
@@ -0,0 +1,103 @@
+from unittest.mock import MagicMock, patch
+import pytest
+import RNS
+from meshchatx.src.backend.auto_propagation_manager import AutoPropagationManager
+
+
+@pytest.mark.asyncio
+async def test_auto_propagation_logic():
+ # Mock dependencies
+ app = MagicMock()
+ context = MagicMock()
+ config = MagicMock()
+ database = MagicMock()
+
+ context.config = config
+ context.database = database
+ context.identity_hash = "test_identity"
+ context.running = True
+
+ manager = AutoPropagationManager(app, context)
+
+ # 1. Test disabled state
+ config.lxmf_preferred_propagation_node_auto_select.get.return_value = False
+ with patch.object(manager, "check_and_update_propagation_node") as mock_check:
+ # Run one iteration manually
+ if config.lxmf_preferred_propagation_node_auto_select.get():
+ await manager.check_and_update_propagation_node()
+ mock_check.assert_not_called()
+
+ # 2. Test selection logic
+ config.lxmf_preferred_propagation_node_auto_select.get.return_value = True
+ config.lxmf_preferred_propagation_node_destination_hash.get.return_value = None
+
+ # Mock announces
+ announce1 = {
+ "destination_hash": "aaaa1111",
+ "app_data": b"\x94\x00\x00\x01\x00", # msgpack for [0, 0, 1, 0] -> enabled=True
+ }
+ announce2 = {"destination_hash": "bbbb2222", "app_data": b"\x94\x00\x00\x01\x00"}
+ database.announces.get_announces.return_value = [announce1, announce2]
+
+ # Mock RNS Transport
+ with (
+ patch.object(RNS.Transport, "has_path", return_value=True),
+ patch.object(RNS.Transport, "hops_to") as mock_hops,
+ patch.object(manager, "probe_node", return_value=True),
+ ):
+ # announce1 is closer (1 hop)
+ # announce2 is further (3 hops)
+ mock_hops.side_effect = lambda dh: 1 if dh == bytes.fromhex("aaaa1111") else 3
+
+ await manager.check_and_update_propagation_node()
+
+ # Should have selected aaaa1111
+ app.set_active_propagation_node.assert_called_with("aaaa1111", context=context)
+ config.lxmf_preferred_propagation_node_destination_hash.set.assert_called_with(
+ "aaaa1111"
+ )
+
+ # 3. Test switching to better node
+ config.lxmf_preferred_propagation_node_destination_hash.get.return_value = (
+ "bbbb2222"
+ )
+ app.set_active_propagation_node.reset_mock()
+
+ with (
+ patch.object(RNS.Transport, "has_path", return_value=True),
+ patch.object(RNS.Transport, "hops_to") as mock_hops,
+ patch.object(manager, "probe_node", return_value=True),
+ ):
+ mock_hops.side_effect = lambda dh: 1 if dh == bytes.fromhex("aaaa1111") else 3
+
+ await manager.check_and_update_propagation_node()
+
+ # Should have switched to aaaa1111 because it's closer
+ app.set_active_propagation_node.assert_called_with("aaaa1111", context=context)
+
+ # 4. Test failover when probe fails
+ config.lxmf_preferred_propagation_node_destination_hash.get.return_value = (
+ "cccc3333"
+ )
+ announce3 = {"destination_hash": "cccc3333", "app_data": b"\x94\x00\x00\x01\x00"}
+ database.announces.get_announces.return_value = [announce1, announce3]
+ app.set_active_propagation_node.reset_mock()
+
+ with (
+ patch.object(RNS.Transport, "has_path", return_value=True),
+ patch.object(RNS.Transport, "hops_to") as mock_hops,
+ patch.object(manager, "probe_node") as mock_probe,
+ ):
+ # announce1 is 1 hop, but probe fails
+ # announce3 is 2 hops, probe succeeds
+ mock_hops.side_effect = lambda dh: 1 if dh == bytes.fromhex("aaaa1111") else 2
+ mock_probe.side_effect = (
+ lambda dh: False if dh == bytes.fromhex("aaaa1111") else True
+ )
+
+ await manager.check_and_update_propagation_node()
+
+ # Should NOT switch to aaaa1111 because probe failed
+ # Should STAY on cccc3333 or switch to it if it was different
+ # Since it's already on cccc3333 and it's the best reachable, no switch
+ app.set_active_propagation_node.assert_not_called()
diff --git a/tests/backend/test_auto_propagation_api.py b/tests/backend/test_auto_propagation_api.py
new file mode 100644
index 00000000..f89d7bee
--- /dev/null
+++ b/tests/backend/test_auto_propagation_api.py
@@ -0,0 +1,101 @@
+import asyncio
+import json
+import shutil
+import tempfile
+from unittest.mock import MagicMock, patch
+import pytest
+import RNS
+from meshchatx.meshchat import ReticulumMeshChat
+
+
+@pytest.fixture
+def temp_dir():
+ dir_path = tempfile.mkdtemp()
+ yield dir_path
+ shutil.rmtree(dir_path)
+
+
+@pytest.fixture
+def mock_rns_minimal():
+ with (
+ patch("RNS.Reticulum") as mock_rns,
+ patch("RNS.Transport"),
+ patch("LXMF.LXMRouter") as mock_lxmf_router,
+ patch("meshchatx.meshchat.get_file_path", return_value="/tmp/mock_path"),
+ patch("meshchatx.meshchat.generate_ssl_certificate"),
+ ):
+ mock_rns_instance = mock_rns.return_value
+ mock_rns_instance.configpath = "/tmp/mock_config"
+ mock_rns_instance.is_connected_to_shared_instance = False
+ mock_rns_instance.transport_enabled.return_value = True
+
+ # Mock LXMF router and its return values to be JSON serializable
+ mock_lxmf_router_instance = mock_lxmf_router.return_value
+ mock_dest = MagicMock()
+ mock_dest.hexhash = "test_lxmf_hexhash"
+ mock_lxmf_router_instance.register_delivery_identity.return_value = mock_dest
+ mock_lxmf_router_instance.propagation_destination = mock_dest
+
+ mock_id = MagicMock(spec=RNS.Identity)
+ mock_id.hash = b"test_hash_32_bytes_long_01234567"
+ mock_id.hexhash = mock_id.hash.hex()
+ mock_id.get_private_key.return_value = b"test_private_key"
+ yield mock_id
+
+
+@pytest.mark.asyncio
+async def test_auto_propagation_api(mock_rns_minimal, temp_dir):
+ app_instance = ReticulumMeshChat(
+ identity=mock_rns_minimal,
+ storage_dir=temp_dir,
+ reticulum_config_dir=temp_dir,
+ )
+
+ # 1. Test GET /api/v1/config includes auto_select
+ get_handler = None
+ for route in app_instance.get_routes():
+ if route.path == "/api/v1/config" and route.method == "GET":
+ get_handler = route.handler
+ break
+
+ assert get_handler is not None
+ request = MagicMock()
+ response = await get_handler(request)
+ data = json.loads(response.body)
+ assert "lxmf_preferred_propagation_node_auto_select" in data["config"]
+ assert data["config"]["lxmf_preferred_propagation_node_auto_select"] is False
+
+ # 2. Test PATCH /api/v1/config updates auto_select
+ patch_handler = None
+ for route in app_instance.get_routes():
+ if route.path == "/api/v1/config" and route.method == "PATCH":
+ patch_handler = route.handler
+ break
+
+ assert patch_handler is not None
+
+ # Update to True
+ mock_request = MagicMock()
+ mock_request.json = MagicMock(return_value=asyncio.Future())
+ mock_request.json.return_value.set_result(
+ {"lxmf_preferred_propagation_node_auto_select": True}
+ )
+
+ response = await patch_handler(mock_request)
+ data = json.loads(response.body)
+ assert data["config"]["lxmf_preferred_propagation_node_auto_select"] is True
+ assert app_instance.config.lxmf_preferred_propagation_node_auto_select.get() is True
+
+ # Update to False
+ mock_request = MagicMock()
+ mock_request.json = MagicMock(return_value=asyncio.Future())
+ mock_request.json.return_value.set_result(
+ {"lxmf_preferred_propagation_node_auto_select": False}
+ )
+
+ response = await patch_handler(mock_request)
+ data = json.loads(response.body)
+ assert data["config"]["lxmf_preferred_propagation_node_auto_select"] is False
+ assert (
+ app_instance.config.lxmf_preferred_propagation_node_auto_select.get() is False
+ )
diff --git a/tests/backend/test_config_manager.py b/tests/backend/test_config_manager.py
index 8d94db1a..df9c147a 100644
--- a/tests/backend/test_config_manager.py
+++ b/tests/backend/test_config_manager.py
@@ -83,3 +83,10 @@ def test_telephony_config(db):
assert config.call_recording_enabled.get() is False
config.call_recording_enabled.set(True)
assert config.call_recording_enabled.get() is True
+
+
+def test_auto_propagation_config(db):
+ config = ConfigManager(db)
+ assert config.lxmf_preferred_propagation_node_auto_select.get() is False
+ config.lxmf_preferred_propagation_node_auto_select.set(True)
+ assert config.lxmf_preferred_propagation_node_auto_select.get() is True
──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────